#pragma once
#include <memory>
#include <mutex>
#include <condition_variable>
#include "muduo/net/TcpConnection.h"
#include "muduo/proto/codec.h"
#include "../comm/Log.h"
#include "../comm/util.h"
#include "../comm/threadpool.h"
#include "../comm/msg.pb.h"
#include "../comm/mq.pb.h"
#include "consumer.h"

// 客户端：在消息队列系统中，客户端是消息的生产者或者消费者。
// 生产者客户端负责创建和发送消息到消息队列中，消费者客户端则负责从消息队列中接收并处理消息。
// 用户：操作客户端(调用客户端提供的函数)，而客户端调用服务器内部对应的函数并将结果返回给用户
namespace ns_channel
{
    using namespace ns_log;
    using namespace ns_util;
    using namespace ns_proto;
    using namespace ns_consume;

    using ProtobufCodecPtr = shared_ptr<ProtobufCodec>;

    // 信道中的函数参数采用协议的方式接收
    using basicCommonResponsePtr = shared_ptr<basicCommonResponse>;
    using basicConsumeResponsePtr = shared_ptr<basicConsumeResponse>;

    // 信道所提供的服务操作
    class Channel
    {
    public:
        using ptr = shared_ptr<Channel>;

        string getCid()
        {
            return _cid;
        }

        Channel(const muduo::net::TcpConnectionPtr &conn,
                const ProtobufCodecPtr &codec, const string &cid)
            : _cid(cid), _conn(conn), _codec(codec) {}

        ~Channel()
        {
            // 如果信道是消费者的话
            basicCancel();
        }

        // 创建、关闭信道(主要进行传递信息给服务端)
        bool createChannel()
        {
            // 1.构建请求
            openChannelRequest req;
            req.set_cid(_cid);
            req.set_rid(Uuid::uuid());

            // 2.将请求发送给服务器(muduo库采用epoll的方式发送数据异步操作)
            _codec->send(_conn, req);

            // 3.等待服务器执行结果并返回
            return waitRsp(req.rid())->ok();
        }
        bool closeChannel()
        {
            // 1.构建请求
            closeChannelRequest req;
            req.set_cid(_cid);
            req.set_rid(Uuid::uuid());

            // 2.将请求发送给服务器(muduo库采用epoll的方式发送数据异步操作)
            _codec->send(_conn, req);

            // 3.等待服务器执行结果并返回
            return waitRsp(req.rid())->ok();
        }

        // 声明、删除交换机
        bool declareExchange(const string &ename, const ExchangeType &type,
                             const bool &IsDurable,
                             const bool &auto_delete,
                             google::protobuf::Map<std::string, std::string> &args)
        {
            // 1.构建请求
            declareExchangeRequest req;
            req.set_cid(_cid);
            req.set_rid(Uuid::uuid());
            req.set_exchange_name(ename);
            req.set_exchange_type(type);
            req.set_durable(IsDurable);
            req.set_auto_delete(auto_delete);
            req.mutable_args()->swap(args);

            // 2.将请求发送给服务器(muduo库采用epoll的方式发送数据异步操作)
            _codec->send(_conn, req);

            // 3.等待服务器执行结果并返回
            return waitRsp(req.rid())->ok();
        }
        bool deleteExchange(const string &ename)
        {
            // 1.构建请求
            deleteExchangeRequest req;
            req.set_cid(_cid);
            req.set_rid(Uuid::uuid());
            req.set_exchange_name(ename);

            // 2.将请求发送给服务器(muduo库采用epoll的方式发送数据异步操作)
            _codec->send(_conn, req);

            // 3.等待服务器执行结果并返回
            return waitRsp(req.rid())->ok();
        }

        // 声明、删除队列
        bool declareQueue(const string &qname,
                          const bool &IsDurable,
                          const bool &exclusive,
                          const bool &auto_delete,
                          google::protobuf::Map<std::string, std::string> &args)
        {
            // 1.构建请求
            declareQueueRequest req;
            req.set_cid(_cid);
            req.set_rid(Uuid::uuid());
            req.set_queue_name(qname);
            req.set_exclusive(exclusive);
            req.set_durable(IsDurable);
            req.set_auto_delete(auto_delete);
            req.mutable_args()->swap(args);

            // 2.将请求发送给服务器(muduo库采用epoll的方式发送数据异步操作)
            _codec->send(_conn, req);

            // 3.等待服务器执行结果并返回
            return waitRsp(req.rid())->ok();
        }
        bool deleteQueue(const string &qname)
        {
            // 1.构建请求
            deleteQueueRequest req;
            req.set_cid(_cid);
            req.set_rid(Uuid::uuid());
            req.set_queue_name(qname);

            // 2.将请求发送给服务器(muduo库采用epoll的方式发送数据异步操作)
            _codec->send(_conn, req);

            // 3.等待服务器执行结果并返回
            return waitRsp(req.rid())->ok();
        }

        // 绑定、解绑交换机与队列信息
        bool queueBind(const string &ename, const string &qname,
                       const string &binding_key)
        {
            // 1.构建请求
            queueBindRequest req;
            req.set_cid(_cid);
            req.set_rid(Uuid::uuid());
            req.set_exchange_name(ename);
            req.set_queue_name(qname);
            req.set_binding_key(binding_key);

            // 2.将请求发送给服务器(muduo库采用epoll的方式发送数据异步操作)
            _codec->send(_conn, req);

            // 3.等待服务器执行结果并返回
            return waitRsp(req.rid())->ok();
        }
        bool queueUnBind(const string &ename, const string &qname)
        {
            // 1.构建请求
            queueUnBindRequest req;
            req.set_cid(_cid);
            req.set_rid(Uuid::uuid());
            req.set_exchange_name(ename);
            req.set_queue_name(qname);

            // 2.将请求发送给服务器(muduo库采用epoll的方式发送数据异步操作)
            _codec->send(_conn, req);

            // 3.等待服务器执行结果并返回
            return waitRsp(req.rid())->ok();
        }

        // 发布消息(给交换机去路由选择队列)-"生产者"
        bool basicPublish(const string &ename,
                          const Attributes *attributes, const string &body)
        {
            // 1.构建请求
            basicPublishRequest req;
            req.set_cid(_cid);
            req.set_rid(Uuid::uuid());
            req.set_exchange_name(ename);
            req.set_body(body);
            if (attributes != nullptr)
            {
                req.mutable_attributes()->set_id(attributes->id());
                req.mutable_attributes()->set_del_mod(attributes->del_mod());
                req.mutable_attributes()->set_routing_key(attributes->routing_key());
            }

            // 2.将请求发送给服务器(muduo库采用epoll的方式发送数据异步操作)
            _codec->send(_conn, req);

            // 3.等待服务器执行结果并返回
            return waitRsp(req.rid())->ok();
        }

        // 消息确认(删除)
        bool basicAck(const string &msgid)
        {
            // *判断是否是消费者信道
            if (_consumer.get() == nullptr)
            {
                LOG(WARNING) << "确认队列消息失败,当前信道不存在消费者" << endl;
                return false;
            }

            // 1.构建请求
            basicAckRequest req;
            req.set_cid(_cid);
            req.set_rid(Uuid::uuid());
            req.set_queue_name(_consumer->_qname);
            req.set_message_id(msgid);

            // 2.将请求发送给服务器(muduo库采用epoll的方式发送数据异步操作)
            _codec->send(_conn, req);

            // 3.等待服务器执行结果并返回
            return waitRsp(req.rid())->ok();
        }

        // 订阅队列消息-"消费者"
        bool basicConsum(const string &consume_tag, const string &qname,
                         const bool &auto_ack, const ConsumerCallback &cb)
        {
            // *构建消费者对象
            if (!_consumer.get())
                _consumer = make_shared<Consumer>(qname, consume_tag, auto_ack, cb);
            else
            {
                LOG(WARNING) << "订阅队列消息失败，当前信道已经订阅过队列消息" << endl;
                return false;
            }

            // 1.构建请求
            basicConsumeRequest req;
            req.set_cid(_cid);
            req.set_rid(Uuid::uuid());
            req.set_consumer_tag(consume_tag);
            req.set_queue_name(qname);
            req.set_auto_ack(auto_ack);

            // 2.将请求发送给服务器(muduo库采用epoll的方式发送数据异步操作)
            _codec->send(_conn, req);

            // 3.等待服务器执行结果并返回
            return waitRsp(req.rid())->ok();
        }

        // 队列消息取消订阅(释放消费者)
        bool basicCancel()
        {
            // *判断是否是消费者信道
            if (_consumer.get() == nullptr)
            {
                LOG(WARNING) << "取消订阅队列消息失败,当前不是消费者信道" << endl;
                return false;
            }

            // 1.构建请求
            basicCancelRequest req;
            req.set_cid(_cid);
            req.set_rid(Uuid::uuid());
            req.set_consumer_tag(_consumer->_tag);

            // 2.将请求发送给服务器(muduo库采用epoll的方式发送数据异步操作)
            _codec->send(_conn, req);

            // *清除消费者
            _consumer.reset();

            // 3.等待服务器执行结果并返回
            return waitRsp(req.rid())->ok();
        }

    public: // 向连接模块提供的接口
        // 连接收到响应(添加)
        bool putBasicRsp(const basicCommonResponsePtr &rsp)
        {
            unique_lock<mutex> lck(_mtx);
            _basic_rsp.insert({rsp->rid(), rsp});
            _cond.notify_all();

            return true;
        }

        // 连接收到消息推送 针对消息的处理操作(用户提供的函数进行回调处理)
        bool consum(const basicConsumeResponsePtr &rsp)
        {
            if (_consumer == nullptr)
            {
                LOG(WARNING) << "消息处理失败,该信道不存在消费者成员对象" << endl;
                return false;
            }
            if (rsp->consumer_tag() != _consumer->_tag)
            {
                LOG(WARNING) << "消息处理失败,客户端与服务器的信道消费者不一致" << endl;
            }
            _consumer->_callback(rsp->consumer_tag(), rsp->mutable_attributes(),
                                 rsp->body());
            return true;
        }

    private:
        basicCommonResponsePtr waitRsp(const string &rid)
        {
            unique_lock<mutex> lck(_mtx);
            _cond.wait(lck, [this, &rid]()
                       { return _basic_rsp.count(rid); });

            basicCommonResponsePtr ret = _basic_rsp[rid];
            _basic_rsp.erase(rid); // 千万别忘了要删除该响应(释放内存)

            return ret;
        }

    private:
        string _cid;                        // 信道标识id
        Consumer::ptr _consumer;            // 信道对应的消费者
        muduo::net::TcpConnectionPtr _conn; // 通信连接(数据发送)
        ProtobufCodecPtr _codec;            // 协议处理器(没必要重复定义，共用)

        // 互斥锁和条件变量保证在发送请求数据成功响应以后才往下继续执行
        mutex _mtx;
        condition_variable _cond;
        // 基础响应的哈希映射判断指定请求是否有响应，有了才满足_cond条件
        unordered_map<string, basicCommonResponsePtr> _basic_rsp;
    };

    // 信道的管理
    class ChannelManage
    {
    public:
        using Ptr = shared_ptr<ChannelManage>;

        ChannelManage() {}

        Channel::ptr createChannel(const muduo::net::TcpConnectionPtr &conn,
                                   const ProtobufCodecPtr &codec)
        {
            // 1.创建信道管理对象
            unique_lock<mutex> lck(_mtx);
            string cid = Uuid::uuid();
            Channel::ptr cp = make_shared<Channel>(conn, codec, cid);
            _channels.insert(make_pair(cid, cp));

            // 返回创建的信道
            return cp;
        }

        bool closeChannel(const string &cid)
        {
            // 1.关闭信道同时释放空间
            unique_lock<mutex> lck(_mtx);
            if (!_channels.count(cid))
            {
                LOG(WARNING) << "关闭信道失败，不存在该id的信道" << endl;
                return false;
            }
            Channel::ptr tmp = _channels[cid];
            _channels.erase(cid);

            return true;
        }

        Channel::ptr getChannel(const string &cid)
        {
            unique_lock<mutex> lck(_mtx);

            if (!_channels.count(cid))
            {
                LOG(WARNING) << "获取信道失败，不存在该id的信道" << endl;
                return Channel::ptr();
            }
            return _channels[cid];
        }

    private:
        mutex _mtx;
        unordered_map<string, Channel::ptr> _channels;
    };

}